package org.example;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import static org.apache.flink.table.api.Expressions.$;

import java.util.Map;

public class HandleJsonExample {
    public static void main(String[] args) throws Exception {
        // 创建 Flink 流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 创建 Table 环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 生成包含JSON字符串的数据
        DataStream<Tuple2<Integer, String>> jsonStream = env.fromElements(
            Tuple2.of(1, "{\"name\":\"张三\",\"age\":30,\"city\":\"北京\"}"),
            Tuple2.of(2, "{\"name\":\"李四\",\"age\":25,\"city\":\"上海\"}"),
            Tuple2.of(3, "{\"name\":\"王五\",\"age\":35,\"city\":\"北京\"}"),
            Tuple2.of(4, "{\"name\":\"赵六\",\"age\":28,\"city\":\"广州\"}"),
            Tuple2.of(5, "{\"name\":\"钱七\",\"age\":32,\"city\":\"上海\"}")
        );

        // 注册为临时表，使用正确的方式
        tableEnv.createTemporaryView("json_table", jsonStream, $("f0").as("id"), $("f1").as("json_data"));

        // 使用JSON_TO_MAP函数将JSON字符串转换为Map结构
        String sql = "SELECT " +
                "id, " +
                "JSON_TO_MAP(json_data) AS json_map, " +
                "JSON_TO_MAP(json_data)['name'] AS name, " +
                "CAST(JSON_TO_MAP(json_data)['age'] AS INT) AS age, " +
                "JSON_TO_MAP(json_data)['city'] AS city " +
                "FROM json_table";
        
        Table resultTable = tableEnv.sqlQuery(sql);
        
        // 执行分组聚合计算
        String aggregateSql = "SELECT " +
                "city, " +
                "COUNT(*) AS person_count, " +
                "AVG(age) AS avg_age " +
                "FROM (" + sql + ") " +
                "GROUP BY city";
        
        Table aggregateTable = tableEnv.sqlQuery(aggregateSql);
        
        // 将聚合结果转换回DataStream并打印，使用toRetractStream处理更新
        tableEnv.toRetractStream(aggregateTable, Row.class).print("城市统计结果");
        
        // 打印原始解析结果
        tableEnv.toDataStream(resultTable).print("原始解析结果");
        
        // 执行作业
        env.execute("JSON处理示例");
    }
}
